package com.atguigu.upp.app;

import com.atguigu.upp.bean.TagInfo;
import com.atguigu.upp.service.MysqlDBService;
import com.atguigu.upp.utils.DBUtil;
import com.atguigu.upp.utils.PropertiesUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.apache.spark.sql.SparkSession;

import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;

/**
 * Created by Smexy on 2022/10/16
 */
public class MergeToWideTableApp
{

    public static void main(String[] args) throws IOException {

        //假设计算的是1号任务，业务日期是2020-06-14
        Integer taskId = Integer.valueOf(args[0]);
        String doDate = args[1];

        //1.查询要计算的task
        SqlSessionFactory sqlSessionFactory = DBUtil.getSqlSessionFactory("mybatis_mysql.xml");
        SqlSession sqlSession = sqlSessionFactory.openSession();
        MysqlDBService mysqlDBService = new MysqlDBService(sqlSession);
        //查询今天要计算的标签有哪些
        List<TagInfo> tags = mysqlDBService.getTagInfoNameEnableToday();

        System.out.println(tags);

        String createTableSql = generateCreateTableSql(tags, doDate);
        String insertSql = generateInsertSql(tags, doDate);

        //3.向表中导入计算的结果
        SparkSession sparkSession = DBUtil.getSparkSession("MergeToWideTableApp");
        sparkSession.sql(createTableSql);
        sparkSession.sql(insertSql);


    }


    /*
create table if not exists upp220608.up_wide_2020_06_14(
    uid string,
    tag_consumer_behavior_order_amount_7d string,
    tag_population_attributes_nature_gender string,
    tag_population_attributes_nature_period string
)
row format delimited fields terminated by '\t'
location ''
     */
    private static String generateCreateTableSql(List<TagInfo> tags,String doDate){

        String template =" create table if not exists %s.%s(" +
                            " uid string,  %s  )" +
                            "row format delimited fields terminated by '\\t' " +
                            "location '%s/%s' ";

        //获取相关的参数
        String db = PropertiesUtil.getProperty("updbname");  //库名
        String table = PropertiesUtil.getProperty("upwideprefix") + doDate.replace('-','_');  //表名
        String path = PropertiesUtil.getProperty("hdfsPath"); //表存储在hdfs的根路径

        //生成动态的列
        List<String> columns = tags
            .stream()
            .map(name -> name.getTagCode().toLowerCase().concat("  string "))
            .collect(Collectors.toList());

        String columnStr = StringUtils.join(columns, ",");

        //格式化
        String createTableSql = String.format(template, db, table, columnStr, path, table);

        System.out.println(createTableSql);

        return createTableSql;
    }

    /*
     select
    *
from (
        select uid,tagValue, 'tag_consumer_behavior_order_amount_7d' tag_code  from upp220608.tag_consumer_behavior_order_amount_7d where dt='2020-06-14'
        union all
        select uid,tagValue, 'tag_population_attributes_nature_gender' tag_code from upp220608.tag_population_attributes_nature_gender where dt='2020-06-14'
        union all
        select uid,tagValue, 'tag_population_attributes_nature_period' tag_code  from upp220608.tag_population_attributes_nature_period where dt='2020-06-14'
         ) t
pivot (
            min(tagValue)
			for tag_code in( 'tag_consumer_behavior_order_amount_7d' ,
			                  'tag_population_attributes_nature_gender',
			                  'tag_population_attributes_nature_period')
       );


     */
    private static String generateInsertSql(List<TagInfo> tags,String doDate){

        String template = " insert overwrite table %s.%s " +
                         "  select * from  ( %s )t pivot ( min(tagValue) for tag_code in ( %s ) ) " ;

        String db = PropertiesUtil.getProperty("updbname");  //库名
        String table = PropertiesUtil.getProperty("upwideprefix") + doDate.replace('-','_');  //表名


        String subQuery = "  select uid,tagValue, '%s' tag_code  from %s.%s where dt='%s' " ;
        //生成 按需查询的列
        List<String> subQuerys = tags
            .stream()
            .map(tag -> String.format(subQuery, tag.getTagCode().toLowerCase(), db, tag.getTagCode().toLowerCase(), doDate))
            .collect(Collectors.toList());

        String subQueryStr = StringUtils.join(subQuerys, "  union all ");

        //生成旋转列值
        List<String> tagColumns = tags
            .stream()
            .map(tag -> "'" + tag.getTagCode().toLowerCase() + "'")
            .collect(Collectors.toList());

        String tagColumnStr = StringUtils.join(tagColumns, ',');

        //最终格式化
        String insertSql = String.format(template, db, table, subQueryStr, tagColumnStr);

        System.out.println(insertSql);


        return insertSql;
    }
}
